package com.zhang.flink.example;

import com.zhang.flink.bean.OrderEvent;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @title: 订单超时检测（keyProcessFunction）版本
 * req：订单在创建之后需要在15分钟内支付
 * @author: zhang
 * @date: 2022/2/13 16:06
 */
public class OrderTimeOut {
    public static void main(String[] args) throws Exception {
        //todo 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<String> result = env
                .readTextFile("src/main/resources/OrderLog.csv")
                .map(new MapFunction<String, OrderEvent>() {
                    @Override
                    public OrderEvent map(String value) throws Exception {
                        String[] fields = value.split(",");
                        return new OrderEvent(
                                fields[0], fields[1], fields[2],
                                Long.parseLong(fields[3]) * 1000L
                        );
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<OrderEvent>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<OrderEvent>() {
                                    @Override
                                    public long extractTimestamp(OrderEvent element, long recordTimestamp) {
                                        return element.ts;
                                    }
                                })
                )
                .keyBy(r -> r.orderId)
                .process(new OrderTimerOutProcess());

        result.print();
        result.getSideOutput(new OutputTag<String>("Payed TimerOut or No Create") {
        }).print("No Create:");
        result.getSideOutput(new OutputTag<String>("Payed TimerOut") {
        }).print("Pay Timeout:");

        //todo 执行任务
        env.execute();
    }

    public static class OrderTimerOutProcess extends KeyedProcessFunction<String, OrderEvent, String> {
        private ValueState<OrderEvent> orderEventState;
        private ValueState<Long> timerTs;

        @Override
        public void open(Configuration parameters) throws Exception {
            orderEventState = getRuntimeContext().getState(
                    new ValueStateDescriptor<OrderEvent>(
                            "create-state", Types.POJO(OrderEvent.class)
                    )
            );

            timerTs = getRuntimeContext().getState(
                    new ValueStateDescriptor<Long>(
                            "timer-ts", Types.LONG
                    )
            );

        }

        @Override
        public void processElement(OrderEvent value, KeyedProcessFunction<String, OrderEvent, String>.Context ctx, Collector<String> out) throws Exception {
            if ("create".equals(value.type)) {
                //如果创建订单，存入状态等待15分钟
                orderEventState.update(value);
                //注册15分钟后的定时器
                long ts = value.ts + 15 * 60 * 1000L;
                ctx.timerService().registerEventTimeTimer(ts);
                //保存定时器时间
                timerTs.update(ts);
            } else if ("pay".equals(value.type)) {
                OrderEvent orderEvent = orderEventState.value();
                if (orderEvent == null) {
                    //由于网络传输，创建订单数据丢失，只有支付数据没有创建数据
                    ctx.output(new OutputTag<String>("Payed TimerOut or No Create") {
                    }, value.orderId + "Payed But No Create");
                } else {
                    //正常支付
                    out.collect(value.orderId + " Create at " + orderEvent.ts + "" +
                            "Payed at " + value.ts);
                    ctx.timerService().deleteEventTimeTimer(timerTs.value());
                    orderEventState.clear();
                    timerTs.clear();
                }
            }
        }

        @Override
        public void onTimer(long timestamp, KeyedProcessFunction<String, OrderEvent, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
            OrderEvent orderEvent = orderEventState.value();
            ctx.output(new OutputTag<String>("Payed TimerOut") {
            }, orderEvent.orderId + " Create But No Pay");
            //清空状态
            timerTs.clear();
            orderEventState.clear();
        }
    }
}
